import collections
import logging
import traceback
import pandas as pd

import coin.proto.coin_order_gateway_pb2 as coin_order_gateway_pb2
from coin.base.timestamp import get_timestamp
from coin.exchange.kr_rest.product.product_impl import generate_product_from_str2
from coin.proto.coin_strategy_pb2 import StrategyLog
from coin.proto.coin_order_gateway_pb2 import OrderGatewayLog, OrderEvent
from coin.support.pta.logic.pta_stats_calculator import get_accounting_currency


class NoOrderEventAlarm():
  MINUTE = 60

  @staticmethod
  def alarm_msg(msg=None):
    return "No Order in last %s min" % NoOrderEventAlarm.MINUTE

class NoStrategyLogAlarm():
  MINUTE = 5

  @staticmethod
  def alarm_msg(msg=None):
    return "No StrategyLog in last %s min" % NoStrategyLogAlarm.MINUTE


class RetiredStrategyAlarm():
  MINUTE = 1

  @staticmethod
  def alarm_msg(msg=None):
    return "Retired but still running"


class NoPnlBalanceAlarm():
  MINUTE = 20

  @staticmethod
  def alarm_msg(msg=None):
    return "No PnlBalance in last %s min" % NoPnlBalanceAlarm.MINUTE


class MultiMachineAlarm():
  MINUTE = 1

  @staticmethod
  def alarm_msg(machines):
    return "Multi machines: %s" % machines


class AbnormalPositionAlarm():
  MINUTE = 5

  @staticmethod
  def alarm_msg(abnormal_positions):
    return "Not in executor config: %s" % abnormal_positions


class AbnormalApplicationAlarm():
  MINUTE = 5

  @staticmethod
  def alarm_msg(msg):
    return "Application Not Found: %s" % msg


class MultiProcessAlarm():
  MINUTE = 1

  @staticmethod
  def alarm_msg(machine):
    return "Multi Process in %s" % machine


class OrderErrorAlarm():
  MINUTE = 5
  FAILED_ORDER_RATIO_THRESHOLD = 0.9
  FAILED_ORDER_NUMBER_THRESHOLD = 10

  @staticmethod
  def alarm_msg(error_ratio_info):
    return "High order error ratio in last 5 min: %s" % error_ratio_info


class StrategyLossAlarm():
  MINUTE = 5
  LOSS_THRESHOLD = -500

  @staticmethod
  def alarm_msg(pnl_in_usd):
    return "Pnl mark in usd less than %s: %.2f" % \
           (StrategyLossAlarm.LOSS_THRESHOLD, pnl_in_usd)


class StrategyStackAlarm():
  MINUTE = 5

  @staticmethod
  def alarm_msg(symbols):
    return "Abnormal stack: %s" % symbols


class TopicNotFoundAlarm():

  @staticmethod
  def alarm_msg(msg=None):
    return "Strategy registered but not launched."


class RateLimitAlarm():
  MINUTE = 30

  @staticmethod
  def alarm_msg(msg):
    return  'Rate limit hit: %s' % msg


def _default_order_info():
  return {'submit_num': 0, 'error_num': 0}


StrategyAlarmStatus = collections.namedtuple('StrategyAlarmStatus',
                                             ['ts', 'alarm_obj', 'key', 'msg'])

StrategyTsStatus = collections.namedtuple(
    'StrategyTsStatus',
    ['current_ts', 'last_log_ts', 'pnl_balance_ts', 'order_event_ts'])

StrategyLogInfo = collections.namedtuple('StrategyLogInfo', ['ts', 'msg', 'type'])


def _no_order_event_alarm_muted(strat_info):
  muted = False
  if strat_info['strategy_group'] in ('blp', 'delta_mgr'):
    muted = True
  if (strat_info['business_unit'] == 'Day1mm' and
      strat_info['strategy_group'] == 'mm_currency_premium'):
     muted = True
  if strat_info['strategy_group'] == 'simple_amm':
    muted = True
  return muted

class StrategyAlarmHandler(object):
  def __init__(self, strategy_info: dict, logger=None):
    self._strategy_info = strategy_info
    self._last_ts = StrategyTsStatus(
        current_ts=0, last_log_ts=0, pnl_balance_ts=0, order_event_ts=0)
    self._total_alarm_msg = collections.defaultdict(list)
    self._logger = logger or logging.getLogger(__name__)
    self._og_position = []
    self._product_config = {}
    self._process_pid_list = []
    self._machine_list = []
    self._order_list = []
    self._account_request_list = []
    self._last_rate_limit = {}
    self._machine_set = set()

  def update_strategy_info(self, strategy_info):
    self._strategy_info.update(strategy_info)

  def update_current_ts(self, ts):
    current_ts = self._last_ts.current_ts
    assert ts >= current_ts, (ts, current_ts)
    self._last_ts = self._last_ts._replace(current_ts=ts)

  def _update_log_ts(self, ts, pb):
    if ts > self._last_ts.last_log_ts:
      self._last_ts = self._last_ts._replace(last_log_ts=ts)
    if pb.HasField('pnl_balance') and ts > self._last_ts.pnl_balance_ts:
      self._last_ts = self._last_ts._replace(pnl_balance_ts=ts)
    if pb.HasField('og_log') and pb.og_log.HasField('event') and \
       ts > self._last_ts.order_event_ts:
      self._last_ts = self._last_ts._replace(order_event_ts=ts)

  def _update_alarm_msg(self, ts, alarm_obj, key, msg):
    strategy_alarm_msg = StrategyAlarmStatus(ts=ts, alarm_obj=alarm_obj, key=key, msg=msg)
    self._total_alarm_msg[alarm_obj].append(strategy_alarm_msg)

  def _update_by_account_request(self, ts, account_request):
    self._account_request_list.append(
        StrategyLogInfo(ts=ts, msg=account_request, type=None))

  def _update_by_position(self, ts, og_log):
    account_position = og_log.position
    self._og_position.append(StrategyLogInfo(ts=ts, msg=og_log, type=StrategyLog.OG_LOG))
    for each_position in account_position.each_position:
      if each_position.HasField('symbol') and each_position.HasField(
          'net_position') and each_position.net_position != 0:
        symbol = each_position.symbol
        config_log = self._product_config.get(symbol)
        if symbol in ['USDT', 'USD']:
          continue
        if config_log and \
           config_log.HasField('lot_size') and config_log.lot_size != 0 and \
           config_log.min_pos != 0 and config_log.max_pos != 0:
          position = each_position.net_position
          if config_log.HasField('reserve'):
            reserve = config_log.reserve
          else:
            reserve = (config_log.max_pos + config_log.min_pos) / 2
          stack = (position - reserve) / config_log.lot_size
          max_stack = (config_log.max_pos - reserve) / config_log.lot_size
          min_stack = (config_log.min_pos - reserve) / config_log.lot_size
          if (stack - max_stack > (max_stack - min_stack) / 2) \
              or (min_stack - stack > (max_stack - min_stack) / 2):
            alarm_msg = {
                symbol: {
                    'pos': position,
                    'max_pos': config_log.max_pos,
                    'min_pos': config_log.min_pos
                }
            }
            if self._strategy_info['strategy_group'] not in ('blp',):
              self._update_alarm_msg(ts, StrategyStackAlarm, symbol, alarm_msg)

  def _update_by_order_event(self, ts, og_log):
    self._order_list.append(StrategyLogInfo(ts=ts, msg=og_log, type=og_log.event.type))

  def _update_strategy_machine(self, ts, machine):
    self._machine_set.add(machine)
    self._machine_list.append(StrategyLogInfo(ts=ts, msg=machine, type=None))

  def _update_by_rate_limit_report_message(self, ts, og_log):
    assert og_log.HasField('rate_limit_report_message'), og_log
    assert og_log.HasField('account_request'), og_log
    rate_limit_type = og_log.rate_limit_report_message.last_hit_request_type
    account_key = (og_log.account_request.market_type,
                   og_log.account_request.exchange,
                   og_log.account_request.owner)
    self._last_rate_limit[(account_key, rate_limit_type)] = StrategyLogInfo(
        ts=ts, msg=og_log, type=StrategyLog.OG_LOG)

  def _handle_og_log(self, ts, pb):
    assert pb.HasField('og_log')
    og_log = pb.og_log
    if og_log.HasField('account_request'):
      self._update_by_account_request(ts, og_log.account_request)
    if og_log.type == OrderGatewayLog.POSITION:
      self._update_by_position(ts, og_log)
    elif og_log.type == OrderGatewayLog.ORDER_EVENT:
      self._update_by_order_event(ts, og_log)
    elif og_log.type == OrderGatewayLog.RATE_LIMIT_REPORT_MESSAGE:
      self._update_by_rate_limit_report_message(ts, og_log)
    else:
      pass

  def _handle_executor_config(self, ts, pb):
    if pb.type == StrategyLog.PASS_EXECUTOR_CONFIG:
      config_log = pb.pass_executor_config
    elif pb.type == StrategyLog.AGG_EXEUCTOR_CONFIG:
      config_log = pb.agg_executor_config
    else:
      return
    self._product_config.update({config_log.symbol: config_log})

  def _handle_process(self, ts, pb, machine):
    self._process_pid_list.append(
        StrategyLogInfo(ts=ts, msg=(pb.process.pid, machine), type=None))

  def _check_application(self):
    account_request_list = [
        account_request_info for account_request_info in self._account_request_list
        if self._last_ts.current_ts - account_request_info.ts < AbnormalApplicationAlarm.MINUTE * 60 * 1e+9
    ]
    self._account_request_list = account_request_list
    if len(account_request_list) == 0:
      return
    unique_key = set()
    for account_request_info in account_request_list:
      account_request = account_request_info.msg
      market_type = account_request.market_type
      exchange = account_request.exchange
      owner = account_request.owner
      if (market_type, exchange, owner) in unique_key:
        continue
      unique_key.add((market_type, exchange, owner))
      ts = account_request_info.ts
      if [account_request.market_type, account_request.exchange, account_request.owner] \
          not in [[account_info.get('market_type'), account_info.get('exchange'),  account_info.get('owner')]
                  for account_info in self._strategy_info['account_info']]:
        self._update_alarm_msg(ts, AbnormalApplicationAlarm, (market_type, exchange, owner),
                               (market_type, exchange, owner))

  def _check_multi_process(self):
    history_pid_info = [
        pid_info for pid_info in self._process_pid_list
        if self._last_ts.current_ts - pid_info.ts < MultiProcessAlarm.MINUTE * 1e+9
    ]
    self._process_pid_list = history_pid_info
    pids = pd.DataFrame([pid_info.msg for pid_info in history_pid_info], columns=['pid', 'machine'])
    if len(pids) == 0:
      return
    multi_machines = []
    for machine in pids['machine'].drop_duplicates():
      pids_unique = pids['pid'][pids['machine'] == machine]
      pids_unique = pids_unique.loc[pids_unique.diff() != 0]
      if len(pids_unique) != len(pids_unique.drop_duplicates()):
        multi_machines.append(machine)
    if multi_machines:
      alarm_msg = ','.join(multi_machines)
      self._update_alarm_msg(self._last_ts.current_ts, MultiProcessAlarm, alarm_msg, alarm_msg)

  def _check_timestamp_alarm(self):
    if not self._strategy_info['active']:
      if (self._last_ts.current_ts - self._last_ts.last_log_ts <
          RetiredStrategyAlarm.MINUTE * 60 * 1e+9):
        self._update_alarm_msg(self._last_ts.current_ts, RetiredStrategyAlarm,
                               self._strategy_info['strategy_name'],
                               self._strategy_info['strategy_name'])
    elif (self._last_ts.current_ts - self._last_ts.last_log_ts >
          NoStrategyLogAlarm.MINUTE * 60 * 1e+9):
      self._update_alarm_msg(self._last_ts.current_ts, NoStrategyLogAlarm,
                             self._strategy_info['strategy_name'],
                             self._strategy_info['strategy_name'])
    elif (self._last_ts.current_ts - self._last_ts.order_event_ts >
          NoOrderEventAlarm.MINUTE * 60 * 1e+9):
      if not _no_order_event_alarm_muted(self._strategy_info):
        self._update_alarm_msg(self._last_ts.current_ts, NoOrderEventAlarm,
                               self._strategy_info['strategy_name'],
                               self._strategy_info['strategy_name'])
    elif (self._last_ts.current_ts - self._last_ts.pnl_balance_ts >
          NoPnlBalanceAlarm.MINUTE * 60 * 1e+9):
      self._update_alarm_msg(self._last_ts.current_ts, NoPnlBalanceAlarm,
                             self._strategy_info['strategy_name'],
                             self._strategy_info['strategy_name'])

  def _check_error_order_ratio(self):
    order_info_list = [
        order_info for order_info in self._order_list
        if self._last_ts.current_ts - order_info.ts < OrderErrorAlarm.MINUTE * 60 * 1e+9
    ]
    self._order_list = order_info_list
    order_status = collections.defaultdict(_default_order_info)
    order_error_msg = collections.defaultdict(dict)
    for order_info in order_info_list:
      og_log=order_info.msg
      market_type = og_log.account_request.market_type
      exchange = og_log.account_request.exchange
      symbol = og_log.event.symbol
      unique_symbol = '%s_%s_%s' % (market_type, exchange, symbol)
      if order_info.type == OrderEvent.ORDER_SUBMITTED:
        order_status[unique_symbol]['submit_num'] += 1
      elif order_info.type in (OrderEvent.ORDER_ERROR, OrderEvent.ORDER_REJECTED):
        order_status[unique_symbol]['error_num'] += 1
        if og_log.event.HasField('exchange_error_code'):
          order_error_msg[unique_symbol].update({og_log.event.exchange_error_code: og_log.event.exchange_error_msg})
        else:
          order_error_msg[unique_symbol].update({'Error code unavaliable': ''})
    for unique_symbol, order_num in order_status.items():
      if order_num['submit_num'] != 0 and order_num[
          'error_num'] > OrderErrorAlarm.FAILED_ORDER_NUMBER_THRESHOLD:
        error_ratio = order_num['error_num'] / order_num['submit_num']
        if error_ratio > OrderErrorAlarm.FAILED_ORDER_RATIO_THRESHOLD:
          symbol = unique_symbol.split('_', 2)[-1]
          self._update_alarm_msg(self._last_ts.current_ts, OrderErrorAlarm, symbol,
                                 str({symbol:order_error_msg[unique_symbol]}))

  def _check_multi_machine(self):
    machine_info_list = [
        machine_info for machine_info in self._machine_list
        if self._last_ts.current_ts - machine_info.ts < MultiMachineAlarm.MINUTE * 1e+9
    ]
    self._machine_list = machine_info_list
    machines = list(set([machine_info.msg for machine_info in machine_info_list]))
    if len(machines) > 1:
      alarm_msg = ','.join(machines)
      self._update_alarm_msg(self._last_ts.current_ts, MultiMachineAlarm, alarm_msg, alarm_msg)

  def _check_abnormal_position(self):
    og_log_with_position = [
        og_log_info for og_log_info in self._og_position
        if self._last_ts.current_ts - og_log_info.ts < AbnormalPositionAlarm.MINUTE * 60 * 1e+9
    ]
    self._og_position = og_log_with_position
    if len(self._product_config) == 0 or len(og_log_with_position) == 0:
      return
    account_request = og_log_with_position[-1].msg.account_request
    account_position = og_log_with_position[-1].msg.position
    position_symbols = [position.symbol for position in account_position.each_position if position.net_position != 0]
    market_type = account_request.market_type
    exchange = account_request.exchange
    config_symbols = [
        generate_product_from_str2(market_type, exchange, None, symbol).symbol
        for symbol in self._product_config.keys()
    ]
    accounting_currency_config = [
        get_accounting_currency(market_type, exchange, symbol) for symbol in config_symbols
    ]
    if market_type == 'Futures' and exchange == 'Okex' and 'USDT' in accounting_currency_config:
      accounting_currency_config.remove('USDT')
    abnormal_position = [
        symbol for symbol in position_symbols if symbol not in config_symbols and
        get_accounting_currency(market_type, exchange, symbol) in accounting_currency_config
    ]
    if len(abnormal_position) > 0:
      self._update_alarm_msg(self._last_ts.current_ts, AbnormalPositionAlarm, abnormal_position, abnormal_position)

  def _check_rate_limit(self):
    rate_limit_info = collections.defaultdict(list)
    for (account_key, rate_limit_type), strat_log_info in \
        self._last_rate_limit.items():
      if rate_limit_type in (coin_order_gateway_pb2.PLACE,
                             coin_order_gateway_pb2.AMEND,
                             coin_order_gateway_pb2.CANCEL):
        if self._last_ts.current_ts - strat_log_info.ts < \
           RateLimitAlarm.MINUTE * 60 * 1e+9:
          assert rate_limit_type not in rate_limit_info[account_key], rate_limit_type
          rate_limit_info[account_key].append(
              coin_order_gateway_pb2.RequestType.Name(rate_limit_type))
    for account_key, rate_limit_types in rate_limit_info.items():
      self._update_alarm_msg(
          self._last_ts.current_ts, RateLimitAlarm,
          account_key,
          '%s: %s' % (account_key, rate_limit_info[account_key]))

  def clean_outdate_alarms(self):
    for alarm_type, strategy_alarm_list in self._total_alarm_msg.items():
      new_alarm_list = []
      for strategy_alarm_status in strategy_alarm_list:
        if self._last_ts.current_ts - strategy_alarm_status.ts < strategy_alarm_status.alarm_obj.MINUTE * 60 * 1e+9:
          new_alarm_list.append(strategy_alarm_status)
      self._total_alarm_msg.update({alarm_type: new_alarm_list})

  def _check_all_alarms(self):
    self._check_timestamp_alarm()
    self._check_error_order_ratio()
    self._check_multi_process()
    self._check_application()
    self._check_multi_machine()
    # TODO(chensili): add it back after multi accounts are supported.
    # self._check_abnormal_position()
    self._check_rate_limit()
    self.clean_outdate_alarms()

  def handle(self, ts, machine, pb):
    try:
      if ts >= self._last_ts.current_ts:
        self.update_current_ts(ts)
      self._update_log_ts(ts, pb)
      if not self._strategy_info['active']:
        return
      if pb.type == StrategyLog.OG_LOG:
        self._handle_og_log(ts, pb)
        self._update_strategy_machine(ts, machine)
      elif pb.type in (StrategyLog.AGG_EXEUCTOR_CONFIG, StrategyLog.PASS_EXECUTOR_CONFIG):
        self._handle_executor_config(ts, pb)
      if pb.HasField('process'):
        self._handle_process(ts, pb, machine)
    except Exception as e:
      self._logger.info(
          'Fail to handle %s' % self._strategy_info['strategy_name'])
      self._logger.info('%s, %s' % (ts, pb))
      self._logger.info(traceback.format_exc())
      raise e


  def get_all_alarms(self):
    self._check_all_alarms()
    all_alarm_msg = {}
    for alarm_type, strategy_alarm_list in self._total_alarm_msg.items():
      key_set = []
      total_msg = []
      for strategy_alarm_status in strategy_alarm_list:
        if strategy_alarm_status.key not in key_set:
          total_msg.append(str(strategy_alarm_status.msg))
          key_set.append(strategy_alarm_status.key)
      if total_msg:
        alarm_msg = strategy_alarm_status.alarm_obj.alarm_msg(','.join(list(self._machine_set) + total_msg))
        all_alarm_msg.update({alarm_type: {self._strategy_info['strategy_name']: alarm_msg}})
    self._total_alarm_msg = collections.defaultdict(list)
    return all_alarm_msg
